package streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming {
  def main(args: Array[String]): Unit = {
    //创建环境对象
    /**StreamingContext创建时，需要传递两个参数:
     * 第一个参数表示环境配置
     * 第二个参数表示批处理的周期（即采集周期）
    */
    val sparkConf=new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    val ssc=new StreamingContext(sparkConf,Seconds(1))
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.80.134", 9999)
    val words: DStream[String] = lines.flatMap(_.split(" "))
    val wordToOne: DStream[(String, Int)] = words.map((_, 1))
    val wordToConut: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)
    wordToConut.print()

    ssc.start()
    ssc.awaitTermination()
//    ssc.stop()
    //关闭环境对象
  }

}
